package net.bwie.jtp.dim.function;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.HConstants;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.util.Bytes;

import java.io.IOException;


/*
  Flink中自定义sink,将DataStreame流中数据写入Hbase表中,要求DataStream数据类型JSON字符串
 */
public class DimHbaseSinkFunction extends RichSinkFunction<String> {


/*

   定义变量:数据库连接

 */

    private Connection connection = null;

    @Override
    public void open(Configuration parameters) throws Exception {
        // TODO: 2025/6/2 创建配置属性,设置属性值,比如连接zk集群信息
        org.apache.hadoop.conf.Configuration conf = HBaseConfiguration.create();
        conf.set(HConstants.CLIENT_ZOOKEEPER_QUORUM, "node101,node102,node103");
        conf.set(HConstants.CLIENT_ZOOKEEPER_CLIENT_PORT, "2181");
        // TODO: 2025/6/2 实例化Connection连接
        connection = ConnectionFactory.createConnection(conf);
    }

    public void invoke(String value, Context context) throws Exception {

        JSONObject jsonObject = JSON.parseObject(value);
        String tableName = jsonObject.getString("hbase_table_name");
        String familyName = jsonObject.getString("family_name");
        String rowKeyColumn = jsonObject.getString("row_key_column");
        String operateType = jsonObject.getString("operate_type");
        JSONObject dataObject = jsonObject.getJSONObject("operate_data");
        // TODO: 2025/6/2 创建表,如果表不存在时:create't1','f1'
        createHbasetable(tableName, familyName);
        // TODO: 2025/6/2 实例化Table对象
        Table table = connection.getTable(TableName.valueOf(tableName));
        // TODO: 2025/6/2 插入数据" put 'dim_base_dic', '1305', 'info:dic_name','拍错'
        if ("insert".equals(operateType) || "update".equals(operateType)) {
            // 依据表的主键字段名称获取对应值
            String rowKeyValue = dataObject.getString(rowKeyColumn);
            Put put = new Put(Bytes.toBytes(rowKeyValue));
            // 添加列
            for (String columnName : dataObject.keySet()) {
                put.addColumn(Bytes.toBytes(familyName),Bytes.toBytes(columnName),Bytes.toBytes(dataObject.getString(columnName)));
                table.put(put);
            }
        }
        // 删除数据:delete 'dim_base_dic' , '1305'
        else if("delete".equals(operateType)){
            // 依据表的主键字段名称获取对应值
            String rowKeyValue = dataObject.getString(rowKeyColumn);
            Delete delete = new Delete(Bytes.toBytes(rowKeyValue));
            table.delete(delete);
        }else {
            System.out.println("操作类型错误,非insert,update和delete类型操作,操作数据"+value);
        }
          table.close();
    }
    /*
      Hbase数据库中表不存在时,依据表名称和列簇名称创建表
     */
    private void createHbasetable(String tableName, String familyName) throws Exception {
        // 管理对象
        Admin admin = connection.getAdmin();
        // 表不存在时,创建表
        TableName tableName1 = TableName.valueOf(tableName);
        if (!admin.tableExists(tableName1)) {
            // 表描述符构造器
            TableDescriptorBuilder tableDescriptorBuilder = TableDescriptorBuilder.newBuilder(tableName1);
            // 列簇描述符构造器
            ColumnFamilyDescriptorBuilder columnFamilyDescriptorBuilder = ColumnFamilyDescriptorBuilder.newBuilder(Bytes.toBytes(familyName));
            // 表添加列簇
            tableDescriptorBuilder.setColumnFamily(columnFamilyDescriptorBuilder.build());
            // 创建表
            admin.createTable(tableDescriptorBuilder.build());
        }
        admin.close();
    }
  @Override
    public void close() throws Exception {
          super.close();
        }
    }


